package cn.xw.DLXQueue;

import cn.xw.utils.ChannelUtil;
import com.rabbitmq.client.Channel;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

/**
 * @author AnHui OuYang
 * @version 1.0
 * created at 2023-04-11 22:14
 * 消费者（正常的消息消费）
 */
public class Consumer {

    //通过日志管理器获取Logger对象
    static Logger logger = LogManager.getLogger(Producer.class);
    //直接交换机名称
    public static final String EXCHANGE_NAME = "MsgHandleExchange";
    //队列名称
    public static final String QUEUE_NAME = "MsgHandleQueue";
    //路由key
    public static final String ROUTING_KEY = "MsgHandleKey";

    //声明死信交换机名称
    public static final String DLX_EXCHANGE = "DLXExchange";
    //声明路由绑定关系 Routing Key 死信交换机到死信队列
    public static final String DLX_KEY = "DLXKey";

    /***
     * 消费者
     */
    public static void main(String[] args) throws IOException {
        //通过工具类获取一个信道
        Channel channel = ChannelUtil.getChannel();

        //声明exchange交换机 并设置直接交换机（路由模式）;防止消费者先启动报错，找不到交换机
        //channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT, true, false, false, null);

        //~~~~~~~~~~~~~~~~~~~~~~~~~~设置死信Start
        //绑定死信队列（参数设置）
        Map<String, Object> arguments = new HashMap<>();
        //正常队列设置死信交换机 参数key是固定值；（就是说死去的消息发送到哪个交换机）
        arguments.put("x-dead-letter-exchange", DLX_EXCHANGE);
        //正常队列设置死信交换机到死信队列绑定Routing Key 参数key是固定值（就是说死去的消息在交换机里通过什么路由发送到死信队列）
        arguments.put("x-dead-letter-routing-key", DLX_KEY);
        //设置正常队列的长度限制 为3
        //arguments.put("x-max-length",3);
        //~~~~~~~~~~~~~~~~~~~~~~~~~~设置死信End

        //创建一个队列
        channel.queueDeclare(QUEUE_NAME, true, false, false, arguments);
        //把队列绑定到指定交换机上
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);

        logger.info("消费者开始监听队列消息....");
        //接收消息
        channel.basicConsume(QUEUE_NAME, false, (consumerTag, message) -> {
            //## 消息处理（处理方式：若处理的消息内包含 "bb" 字符串则拒绝处理，并设置手动不确认，并丢弃消息 ）
            //获取消息体
            String msg = new String(message.getBody(), StandardCharsets.UTF_8);
            logger.info("消费者获取队列信息正在处理：{}", msg);
            if (msg.contains("bb")) {
                //拒绝处理
                logger.info("=======》 X 消息已被拒绝处理，并丢弃：{}", msg);
                channel.basicNack(message.getEnvelope().getDeliveryTag(), false, false);
            } else {
                logger.info("=======》 √ 消息已被成功处理：{}", msg);
                channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            }
        }, consumerTag -> {
            logger.info("监听的队列出现异常；可能队列被删除！{}", consumerTag);
        });
    }
}